package com.codejiwei.flink.udfs;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;


/**
 * author: codejiwei
 * date: 2023/6/7
 * desc:
 **/
public class Flink_UDF_Test {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(5);
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        tEnv.createTemporarySystemFunction("json_extract", JsonExtractFunction.class);

        DataStreamSource<String> fileDS = env.readTextFile("src/main/resources/data/json.data");


        Schema schema = Schema.newBuilder()
                .column("f0", DataTypes.STRING())
                .build();
        Table table1 = tEnv.fromDataStream(fileDS, schema);
        tEnv.createTemporaryView("t_view", table1);


//        Table table = tEnv.sqlQuery("select json_extract(f0, '$.address.country') from t_view");
//        table.execute().print();



        tEnv.executeSql("create table sinkTable with ('connector'='print') like t_view");
        tEnv.executeSql("insert into sinkTable from select * from t_view");








        //输出
//        tEnv.createTemporaryTable("SourceTable", TableDescriptor.forConnector("datagen")
//                .schema(Schema.newBuilder()
//                        .column("f0", DataTypes.STRING())
//                        .build())
//                .option(DataGenConnectorOptions.ROWS_PER_SECOND, 10L)
//                .build());
//
//        tEnv.executeSql("CREATE TEMPORARY TABLE SinkTable WITH ('connector' = 'print') LIKE SourceTable (EXCLUDING OPTIONS) ");


//        tEnv.executeSql("insert into SinkTable select * from SourceTable");


//        Table table1 = tEnv.from("SourceTable");
//        table1.executeInsert("SinkTable");
//
//
//        Table table2 = tEnv.sqlQuery("SELECT * FROM SourceTable");
//        table2.execute().print();



        //创建表
        //创建视图
//        tEnv.createTemporaryView("t_view", fileDS);
//        tEnv.createTemporaryView("t_view", table);
//        tEnv.sqlQuery("select * from t_view").execute().print();



//        tEnv.createTemporaryTable("");
//        tEnv.createTemporaryView("table1", table1);
//        tEnv.executeSql("create table sinkTable with ('connector' = 'print') like table1");
//        table1.executeInsert("sinkTable");


//
//        DataStreamSource<String> fileDS = env.readTextFile("src/main/resources/data/json.data");
//
//        tEnv.createTemporaryTable("SourceTable", TableDescriptor.forConnector("datagen")
//                .schema(Schema.newBuilder()
//                        .column("f0", DataTypes.STRING())
//                        .build())
//                .option(DataGenConnectorOptions.ROWS_PER_SECOND, 100L)
//                .build());
//
//
//
//
//        tEnv.executeSql("CREATE TEMPORARY TABLE SinkTable WITH ('connector' = 'blackhole') " +
//                "LIKE SourceTable (EXCLUDING OPTIONS) ");
//
//
//        Table table1 = tEnv.from("SourceTable");
////
//        Table table2 = tEnv.sqlQuery("SELECT * FROM SourceTable");
////        TableResult tableResult = table1.insertInto("SinkTable").execute();
//        tEnv.toDataStream(table1).print();






//        tEnv.createTemporaryView("table1", fileDS);



//        Schema schema = Schema.newBuilder()
////                .column("field1", DataTypes.STRING())
////                .build();
//            Table table1 = tEnv.fromDataStream(fileDS, $("field1"));
//
////        tEnv.createTemporaryView("table1", fileDS, schema);
//
//
//        Table resultTable = tEnv.sqlQuery("select json_extract(field1, $.address) from " + table1);
//        tEnv.toDataStream(resultTable, Row.class).print();

        env.execute();

    }
}
